package io.kiki.sba.registry.server.data_store.resource;


import io.kiki.sba.registry.api.Server;
import io.kiki.sba.registry.api.exchange.DataExchanger;
import io.kiki.sba.registry.common.model.ClientToClientInterfaceServerNodeId;
import io.kiki.sba.registry.common.model.constants.ValueConstants;
import io.kiki.sba.registry.common.model.dataserver.Datum;
import io.kiki.sba.registry.common.model.store.DataInfo;
import io.kiki.sba.registry.common.model.store.Publisher;
import io.kiki.sba.registry.remoting.netty.Channel;
import io.kiki.sba.registry.server.data_store.bootstrap.ServerConfig;
import io.kiki.sba.registry.server.data_store.cache.DatumStorageDelegate;
import io.kiki.sba.registry.server.data_store.remoting.metaserver.MetaServerServiceImpl;
import io.kiki.sba.registry.util.StringUtil;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import javax.ws.rs.*;
import javax.ws.rs.core.MediaType;
import java.util.*;
import java.util.Map.Entry;


@Path("digest")
public class DataDigestResource {
    private static final Logger logger = LoggerFactory.getLogger(DataDigestResource.class);

    private static final String SESSION = "SESSION";

    private static final String META = "META";

    @Autowired
    DataExchanger boltDataExchanger;

    @Autowired
    MetaServerServiceImpl metaServerService;

    @Autowired
    ServerConfig serverConfig;

    @Autowired
    DatumStorageDelegate datumStorageDelegate;

    @GET
    @Path("datum/query")
    @Produces(MediaType.APPLICATION_JSON)
    public Map<String, Datum> getDatumByDataInfoId(@QueryParam("dataId") String dataId, @QueryParam("group") String group, @QueryParam("instanceId") String instanceId, @QueryParam("dataCenter") String dataCenter) {

        StringUtil.checkNotBlank(dataId, "dataId");
        StringUtil.checkNotBlank(group, "group");
        if (StringUtils.isBlank(instanceId)) {
            instanceId = ValueConstants.DEFAULT_INSTANCE_ID;
        }
        if (StringUtils.isBlank(dataCenter)) {
            dataCenter = serverConfig.getDataCenter();
        }
        String dataInfoId = DataInfo.toDataInfoId(instanceId, dataId, group);
        Map<String, Datum> retList = new HashMap<>();
        retList.put(dataCenter, datumStorageDelegate.get(dataCenter, dataInfoId));
        return retList;
    }

    @POST
    @Path("connect/query")
    @Produces(MediaType.APPLICATION_JSON)
    public Map<String, Map<String, Publisher>> getPublishersByConnectId(Map<String, String> map) {
        Map<String, Map<String, Publisher>> ret = new HashMap<>();
        if (map != null && !map.isEmpty()) {
            map.forEach((clientConnectId, sessionConnectId) -> {
                ClientToClientInterfaceServerNodeId clientToClientInterfaceServerNodeId = ClientToClientInterfaceServerNodeId.of(clientConnectId, sessionConnectId);
                Map<String, Publisher> publisherMap = datumStorageDelegate.getByConnectId(clientToClientInterfaceServerNodeId);
                if (publisherMap != null && !publisherMap.isEmpty()) {
                    ret.put(clientToClientInterfaceServerNodeId.toString(), publisherMap);
                }
            });
        }
        return ret;
    }

    @GET
    @Path("datum/count")
    @Produces(MediaType.APPLICATION_JSON)
    public String getDatumCount() {
        return getLocalDatumCount();
    }

    protected String getLocalDatumCount() {
        StringBuilder sb = new StringBuilder("CacheDigest");
        try {

            Map<String, Map<String, Datum>> allMap = datumStorageDelegate.getLocalAll();
            if (!allMap.isEmpty()) {
                for (Entry<String, Map<String, Datum>> dataCenterEntry : allMap.entrySet()) {
                    String dataCenter = dataCenterEntry.getKey();
                    Map<String, Datum> datumMap = dataCenterEntry.getValue();
                    sb.append(String.format(" [Datum] size of datum in %s is %s", dataCenter, datumMap.size()));
                    int pubCount = datumMap.values().stream().mapToInt(Datum::publisherSize).sum();
                    sb.append(String.format(",[Publisher] size of publisher in %s is %s", dataCenter, pubCount));
                }
            } else {
                sb.append(" datum cache is empty");
            }

        } catch (Throwable t) {
            sb.append(" cache digest error!");
        }

        return sb.toString();
    }

    @GET
    @Path("{type}/serverList/query")
    @Produces(MediaType.APPLICATION_JSON)
    public Map<String, List<String>> getServerListAll(@PathParam("type") String type) {
        Map<String, List<String>> map = new HashMap<>();
        if (StringUtils.isNotBlank(type)) {
            String inputType = type.trim().toUpperCase();

            switch (inputType) {
            case SESSION:
                List<String> sessionList = getSessionServerList();
                map.put(serverConfig.getDataCenter(), sessionList);
                break;
            case META:
                List<String> metaList = getMetaServerLeader();
                map.put(serverConfig.getDataCenter(), metaList);
                break;
            default:
                throw new IllegalArgumentException("unsupported server type:" + type);
            }
        }

        return map;
    }

    List<String> getSessionServerList() {
        Server server = boltDataExchanger.getServer(serverConfig.getPort());
        if (server == null) {
            return Collections.emptyList();
        }
        Map<String, io.kiki.sba.registry.api.Channel> channels = server.selectAvailableChannelsForHostAddress();
        List<String> ret = new ArrayList<>(channels.size());
        for (io.kiki.sba.registry.api.Channel channel : channels.values()) {
            Channel boltChannel = (Channel) channel;
            io.kiki.stack.netty.channelManager.Channel conn = boltChannel.getChannel();
            ret.add(conn.getRemoteIp() + ":" + conn.getRemotePort());
        }
        return ret;
    }

    List<String> getMetaServerLeader() {
        String leader = metaServerService.getMetaServerLeader();
        if (StringUtils.isNotBlank(leader)) {
            return new ArrayList<>(Arrays.asList(leader));
        } else {
            return Collections.emptyList();
        }
    }

    @GET
    @Path("datum/getDataInfoIdList")
    @Produces(MediaType.APPLICATION_JSON)
    public Collection<String> getDataInfoIdList() {
        try {
            Map<String, Map<String, Datum>> allMap = datumStorageDelegate.getLocalAll();
            if (CollectionUtils.isEmpty(allMap)) {
                return Collections.emptyList();
            }
            List<String> dataInfos = new ArrayList<>(1024);
            StringBuilder builder = new StringBuilder(128);
            for (Entry<String, Map<String, Datum>> dataCenterEntry : allMap.entrySet()) {
                String dataCenter = dataCenterEntry.getKey();
                Map<String, Datum> datumMap = dataCenterEntry.getValue();
                for (Datum datum : datumMap.values()) {
                    builder.append(dataCenter).append(",").append(datum.getDataInfoId()).append(",").append(datum.publisherSize());

                    dataInfos.add(builder.toString());
                    builder.setLength(0);
                }
            }
            return dataInfos;
        } catch (Throwable e) {
            logger.error("failed to get dataInfoList", e);
            throw new RuntimeException(e);
        }
    }
}
